/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nageoffer.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessLogsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkBrowserStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkDeviceStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkLocaleStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkNetworkStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkOsStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkStatsTodayDO;
import com.nageoffer.shortlink.project.dao.entity.ShortLinkGotoDO;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessLogsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkBrowserStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkDeviceStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkLocaleStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkNetworkStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkOsStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkStatsTodayMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkGotoMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkMapper;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.nageoffer.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.nageoffer.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

// Redis Stream消息消费者实现类
// 负责处理短链接访问统计数据的消费逻辑
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    // 数据库Mapper注入
    private final ShortLinkMapper shortLinkMapper;  // 短链接基础信息Mapper
    private final ShortLinkGotoMapper shortLinkGotoMapper;  // 短链接路由Mapper
    private final RedissonClient redissonClient;  // 分布式锁客户端
    private final LinkAccessStatsMapper linkAccessStatsMapper;  // 访问统计Mapper
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;  // 地域统计Mapper
    private final LinkOsStatsMapper linkOsStatsMapper;  // 操作系统统计Mapper
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;  // 浏览器统计Mapper
    private final LinkAccessLogsMapper linkAccessLogsMapper;  // 访问日志Mapper
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;  // 设备统计Mapper
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;  // 网络统计Mapper
    private final LinkStatsTodayMapper linkStatsTodayMapper;  // 今日统计Mapper
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;  // 延迟消息生产者
    private final StringRedisTemplate stringRedisTemplate;  // Redis操作模板
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;


    // 高德地图API密钥配置
    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;



    //正常情况：获取消息 -> 设置预占标识（无，设置正在消费中） -> 消费 -> 标识设置（已完成）
    //异常情况：
    //1.重复消息：重复消息 -> 判断预占标识（正在消费中） -> 判断是否消费完成（已完成）-> 返回
    //2.重试消息（正常流程）：重试消息 -> 判断预占标识（无，设置正在消费中）->消费 ->标识设置（已完成）
    //3.重试消息（之前设置已完成标识失败）：重试消息 -> 判断预占标识（正在消费中）->  判断是否消费完成（未完成）-> 抛出异常 消息消费中 但是未完成 -> 重试 ->redis过期 -> 进入正常流程
    //4.正常消息（消费异常）：获取消息 -> 判断预占标识（无，设置正在消费中）-> 消费异常 ->删除预占标识  -> 抛出异常 消息消费失败 ->重试
    //5.重试消息（之前消费异常，并且删除预占标识失败“宕机”）：参考第3条


    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // 获取消息所属的Stream名称和消息ID
        String stream = message.getStream();
        RecordId id = message.getId();

        // 检查当前消息是否已被处理（通过Redis幂等键判断）
        if (!messageQueueIdempotentHandler.isMessageProcessed(id.toString())) {
            // 若未被处理，则检查消息流程是否已完成（防止重复消费）
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return; // 已完成则直接返回，避免重复处理
            }
            // 未完成流程则抛出异常，触发消息队列重试机制
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }

        try {
            // 解析消息体中的生产者数据
            Map<String, String> producerMap = message.getValue();
            String fullShortUrl = producerMap.get("fullShortUrl");

            // 校验短链接有效性并处理统计逻辑
            if (StrUtil.isNotBlank(fullShortUrl)) {
                String gid = producerMap.get("gid");
                // 反序列化统计记录DTO
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(
                        producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                // 实际执行短链接统计存储逻辑
                actualSaveShortLinkStats(fullShortUrl, gid, statsRecord);
            }

            // 处理成功后删除Redis Stream中的消息（ACK机制替代）
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex) {
            // 捕获所有异常，清除幂等标记以允许消息重试
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            // 记录错误日志（包含异常堆栈）
            log.error("记录短链接监控消费异常", ex);
        }

        // 无论成功与否，最终标记消息流程完成（防止无限重试）
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }


    // 实际统计逻辑实现
    public void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {
        // 参数处理：优先使用方法参数中的fullShortUrl
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());

        // 获取分布式读写锁（防止分组信息并发修改）
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();

        // 尝试获取读锁（非阻塞）
        if (!rLock.tryLock()) {
            // 获取锁失败时发送延迟消息重试
            delayShortLinkStatsProducer.send(statsRecord);
            return;
        }

        try {
            // 处理分组ID（gid）
            if (StrUtil.isBlank(gid)) {
                // 查询路由表获取分组ID
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }

            // 获取当前时间信息
            int hour = DateUtil.hour(new Date(), true);  // 当前小时（24小时制）
            Week week = DateUtil.dayOfWeekEnum(new Date());  // 当前星期
            int weekValue = week.getIso8601Value();  // ISO标准星期值（1-7）

            // 构建基础访问统计记录
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)  // 页面访问量+1
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)  // 根据UV首次标记决定是否计数
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)  // 根据UIP首次标记决定是否计数
                    .hour(hour)  // 当前小时
                    .weekday(weekValue)  // 当前星期
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .gid(gid)  // 分组ID
                    .date(new Date())  // 当前日期
                    .build();
            // 写入基础访问统计
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);

            // 地理位置统计处理
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);  // 高德地图API密钥
            localeParamMap.put("ip", statsRecord.getRemoteAddr());  // 客户端IP
            // 调用高德地图API获取地理位置
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);
            String infoCode = localeResultObj.getString("infocode");

            // 默认地理位置值
            String actualProvince = "未知";
            String actualCity = "未知";

            // 成功获取地理位置时的处理
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                String province = localeResultObj.getString("province");
                boolean unknownFlag = StrUtil.equals(province, "[]");
                // 构建地理位置统计记录
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))
                        .cnt(1)  // 计数+1
                        .fullShortUrl(fullShortUrl)  // 完整短链接
                        .country("中国")  // 默认国家
                        .gid(gid)  // 分组ID
                        .date(new Date())  // 当前日期
                        .build();
                // 写入地理位置统计
                linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
            }

            // 操作系统统计
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())  // 客户端操作系统
                    .cnt(1)  // 计数+1
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 当前日期
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);

            // 浏览器统计
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())  // 客户端浏览器
                    .cnt(1)  // 计数+1
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 当前日期
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);

            // 设备统计
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())  // 客户端设备
                    .cnt(1)  // 计数+1
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 当前日期
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            // 网络类型统计
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())  // 客户端网络类型
                    .cnt(1)  // 计数+1
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 当前日期
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            // 访问日志记录
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())  // UV标识
                    .ip(statsRecord.getRemoteAddr())  // 客户端IP
                    .browser(statsRecord.getBrowser())  // 浏览器
                    .os(statsRecord.getOs())  // 操作系统
                    .network(statsRecord.getNetwork())  // 网络类型
                    .device(statsRecord.getDevice())  // 设备类型
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))  // 地理位置
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            // 更新基础统计计数器
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1,
                    statsRecord.getUvFirstFlag() ? 1 : 0,
                    statsRecord.getUipFirstFlag() ? 1 : 0);

            // 今日统计记录
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)  // 今日PV+1
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)  // 根据UV标记决定今日UV是否+1
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)  // 根据UIP标记决定今日UIP是否+1
                    .gid(gid)  // 分组ID
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 当前日期
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } catch (Throwable ex) {
            // 统计异常处理
            log.error("短链接访问量统计异常", ex);
        } finally {
            // 确保锁释放
            rLock.unlock();
        }
    }
}